package cd.zhixuan.utils.task;

import androidx.annotation.IntRange;
import androidx.annotation.NonNull;
import androidx.annotation.Nullable;

import java.io.OutputStream;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

/**
 * 写入输出流的Task
 *
 * @author: HuZC
 * @Description:
 * @Date: 2022/1/11
 */
public class OutputStreamTask {
    @Nullable
    private OnCallBack onCallBack;
    @NonNull
    private final OutputStream outputStream;
    @Nullable
    private Disposable writeDisposable;
    @NonNull
    private final LinkedBlockingQueue<byte[]> writeQueue;
    /**
     * 释放资源超时时间
     */
    @IntRange(from = 0, to = Integer.MAX_VALUE)
    public int releaseTimeout = 30000;//毫秒

    public OutputStreamTask(@NonNull OutputStream os) {
        this(os, 16);
    }

    /**
     * @param os       OutputStream
     * @param capacity 任务栈大小
     */
    public OutputStreamTask(@NonNull OutputStream os, @IntRange(from = 1, to = Integer.MAX_VALUE) int capacity) {
        outputStream = os;
        writeQueue = new LinkedBlockingQueue<>(capacity);
    }

    public void write(byte[] data) {
        if (data == null) {
            data = new byte[0];
        }
        try {
            if (onCallBack != null) {
                onCallBack.preWrite(data);
            }
            writeQueue.put(data);
        } catch (Exception e) {
            if (onCallBack != null) {
                onCallBack.error(data, e);
            }
        }
        if (writeDisposable != null && !writeDisposable.isDisposed()) {
            return;
        }
        writeDisposable = Observable.just("")
                .observeOn(Schedulers.io())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        long startTime = System.currentTimeMillis();
                        long currentTime;
                        long pollTimeout = releaseTimeout / 10;//十分之一的时间获取超时
                        byte[] bytes;
                        while (true) {
                            try {
                                bytes = writeQueue.poll(pollTimeout, TimeUnit.MILLISECONDS);
                            } catch (Exception e) {
                                bytes = null;
                                if (onCallBack != null) {
                                    onCallBack.error(null, e);
                                }
                            }
                            if (bytes != null) {
                                startTime = System.currentTimeMillis();//刷新时间
                                pollTimeout = releaseTimeout / 10;
                                try {
                                    outputStream.write(bytes);
                                    if (onCallBack != null) {
                                        onCallBack.writed(bytes);
                                    }
                                } catch (Exception e) {
                                    if (onCallBack != null) {
                                        onCallBack.error(bytes, e);
                                    }
                                }
                            } else {
                                currentTime = System.currentTimeMillis();
                                long temp = currentTime - startTime;
                                if (temp >= releaseTimeout) {
                                    //超时
                                    break;
                                } else {
                                    //离超时释放时间少于获取元素的超时时间，重新设置获取元素的超时时间
                                    temp = releaseTimeout - temp;
                                    if (temp < pollTimeout) {
                                        pollTimeout = temp;
                                    }
                                }
                            }

                        }
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        throwable.printStackTrace();
                    }
                });
    }

    public void stopWrite() {
        if (writeDisposable != null) {
            writeDisposable.dispose();
            writeDisposable = null;
        }
        writeQueue.clear();
    }

    public boolean isWriting() {
        return writeDisposable != null && !writeDisposable.isDisposed();
    }

    public void setOnCallBack(OnCallBack onCallBack) {
        this.onCallBack = onCallBack;
    }


    public interface OnCallBack {
        void preWrite(@NonNull byte[] data);

        void writed(@NonNull byte[] data);

        void error(@Nullable byte[] data, @NonNull Exception e);
    }
}
